/**
 * 1) 同步/异步发送：ActiveMQConnectionFactory.setUseAsyncSend(true);异步发送通过回调确认是否发送成功
 * 2）消息事务：JmsTemplate.setSessionTransacted(true);
 * 3）消息持久化：JmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT)
 * 4）消息确认：JmsTemplate.setSessionTransacted(false)+JmsTemplate.setSessionAcknowledgeMode(ActiveMQSession.CLIENT_ACKNOWLEDGE)
 * 5）死信：持久化消息重发送次数超过设定值或过期会被送到死信队列（默认队列名为：ActiveMQ.DLQ）
 * 6）延迟发送：JmsTemplate.send()中使用MessagePostProcessor参数，在其中设置延迟时间message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, delay)
 *
 * Spring提供了两个javax.jms.ConnectionFactory的实现：SingleConnectionFactory（仅缓存connection和session）和CachingConnectionFactory（额外缓存producer和consumer）
 * org.apache.activemq.pool.PooledConnectionFactory会引入一个常用的PooledConnectionFactory类，会缓存connection，session，producer
 *
 * SimpleMessageConverter是MessageConverter的默认实现。用来在Java对象与JMS消息之间进行转换可使用它们分别将String转换为JMS TextMessage，
 * 字节数组(byte[])转换为JMS BytesMessage，Map转换为JMS MapMessage，Serializable对象转换为JMS ObjectMessage
 */
package com.example.acitvemqspringbootdemo;

import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.jms.connection.JmsTransactionManager;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.SimpleMessageListenerContainer;
import org.springframework.jms.listener.adapter.MessageListenerAdapter;
import org.springframework.transaction.TransactionManager;
import org.springframework.transaction.annotation.EnableTransactionManagement;
import org.springframework.transaction.annotation.Transactional;

import javax.jms.*;

@Configuration
@EnableJms  //触发@JmsListener，可以用该注释方法异步接收消息
public class QueueConfig {
    @Value("${queue}")
    private String queue;

    @Bean
    public Queue myQueue() {
        return new ActiveMQQueue(queue);
    }

    @Bean
    public Queue sendQueue() {
        return new ActiveMQQueue("send_reply_queue");
    }

    @Bean
    public Queue replyQueue() {
        return new ActiveMQQueue("reply_queue");
    }

    @Bean
    public Topic myTopic() {
        return new ActiveMQTopic("myTopic");
    }

    @Bean
    public RedeliveryPolicy redeliveryPolicy() {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        //是否在每次尝试重新发送失败后,增长这个等待时间
        redeliveryPolicy.setUseExponentialBackOff(true);
        //重发次数,默认为6次   这里设置为3次
        redeliveryPolicy.setMaximumRedeliveries(3);
        //重发时间间隔,默认为1秒
        redeliveryPolicy.setInitialRedeliveryDelay(1);
        //第一次失败后重新发送之前等待500毫秒,第二次失败再等待500 * 2毫秒,这里的2就是value
        redeliveryPolicy.setBackOffMultiplier(2);
        //是否避免消息碰撞
        redeliveryPolicy.setUseCollisionAvoidance(false);
        //设置重发最大拖延时间-1 表示没有拖延只有UseExponentialBackOff(true)为true时生效
        redeliveryPolicy.setMaximumRedeliveryDelay(-1);
        return redeliveryPolicy;
    }

    //标准的ConnectionFactory
    @Primary
    @Bean
    public ConnectionFactory connectionFactory(@Value("${spring.activemq.broker-url}") String url) {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("admin", "admin", url);
        //启用缓存
        return new CachingConnectionFactory(activeMQConnectionFactory);
    }

    //异步发送的ConnectionFactory
    @Bean
    public ActiveMQConnectionFactory activeMQConnectionFactoryAsync(@Value("${spring.activemq.broker-url}") String url, RedeliveryPolicy redeliveryPolicy) {
        ActiveMQConnectionFactory activeMQConnectionFactory =
                new ActiveMQConnectionFactory(
                        "admin",
                        "admin",
                        url);

        //异步发送，AlwaysSyncSend为true则全部是同步发送，为false时NON_PERSISTENT”、事务中的消息将使用“异步发送"，PERSISTENT消息
        //为同步发送，此时UseAsyncSend为true时，PERSISTENT消息则为异步发送；若jms.sendTimeout>0将会忽略(alwaysSyncSend、useAsyncSend和消息是否持久化)
        //所有的消息都是用同步发送
        activeMQConnectionFactory.setAlwaysSyncSend(false);    //默认false
        activeMQConnectionFactory.setUseAsyncSend(true);       //默认false
        activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        return activeMQConnectionFactory;
    }

    @Bean
    public JmsMessagingTemplate jmsMessagingTemplate(ConnectionFactory connectionFactory) {
        return new JmsMessagingTemplate(connectionFactory);
    }

    //标准的JmsTemplate
    @Bean
    public JmsTemplate jmsTemplate(ActiveMQConnectionFactory connectionFactory) {
        JmsTemplate jmsTemplate = new JmsTemplate();
        jmsTemplate.setConnectionFactory(connectionFactory);
        return jmsTemplate;
    }

    //异步发送JmsTemplate
    @Bean
    public JmsTemplate jmsTemplate_async(ActiveMQConnectionFactory activeMQConnectionFactoryAsync) {
        JmsTemplate jmsTemplate = new JmsTemplate();
        jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);
        jmsTemplate.setConnectionFactory(activeMQConnectionFactoryAsync);
        return jmsTemplate;
    }
    //关闭事务且需要消息确认的JmsTemplate；默认是开启事务处理的，消息确认机制与之是冲突的
    @Bean
    public JmsTemplate jmsTemplate_ack(ActiveMQConnectionFactory activeMQConnectionFactoryAsync) {
        JmsTemplate jmsTemplate = new JmsTemplate();
        //关闭事务
        jmsTemplate.setSessionTransacted(false);
        //开启消息手动确认（发送端）
        jmsTemplate.setSessionAcknowledgeMode(ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);  //客户端签收模式
        jmsTemplate.setConnectionFactory(activeMQConnectionFactoryAsync);
        return jmsTemplate;
    }

    @Bean
    JmsTransactionManager transactionManager(ConnectionFactory connectionFactory) {
        return new JmsTransactionManager(connectionFactory);
    }

    @Bean(name = "jmsTransactionListener")
    public JmsListenerContainerFactory<DefaultMessageListenerContainer> transactionFactory(ActiveMQConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        //开启事务处理
        factory.setSessionTransacted(true);   //默认为false
//替代        factory.setTransactionManager(transactionManager(connectionFactory));
//        factory.setSessionAcknowledgeMode(4);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }

    @Bean(name = "jmsAckListener")
    public JmsListenerContainerFactory<SimpleMessageListenerContainer> noTransactionFactory(ActiveMQConnectionFactory activeMQConnectionFactoryAsync) {
        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
        //关闭事务处理
        factory.setSessionTransacted(false);
        //开启消息手动确认（接收端）
        factory.setSessionAcknowledgeMode(4);
        factory.setConnectionFactory(activeMQConnectionFactoryAsync);
        return factory;
    }

    //定义一个消息监听器连接工厂，这里定义的是点对点模式的监听器连接工厂
    @Bean(name = "jmsQueueListener")
    public JmsListenerContainerFactory<DefaultMessageListenerContainer> jmsQueueListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置连接数
        factory.setConcurrency("1-10");
        //重连间隔时间
        factory.setRecoveryInterval(1000L);

        //设置是否Topic
        factory.setPubSubDomain(false);    //Queue监听

        return factory;
    }

    @Bean("jmsTopicListener")
    public JmsListenerContainerFactory<DefaultMessageListenerContainer> jmsTopicListenerContainerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        //设置是否Topic
        factory.setPubSubDomain(true);   //Topic监听
        return factory;
    }

    //持久化订阅
    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactoryTopicDurable(ActiveMQConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(activeMQConnectionFactory);
        factory.setPubSubDomain(true);
        //开启持久化订阅
        factory.setSubscriptionDurable(true);
        return factory;
    }

    @Bean(name = "topicContainerFactory1")
    public DefaultJmsListenerContainerFactory topicClient1(ActiveMQConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        //开启持久化订阅
        factory.setSubscriptionDurable(true);
        factory.setClientId("1001");
        return factory;
    }

    @Bean(name = "topicContainerFactory2")
    public DefaultJmsListenerContainerFactory topicClient2(ActiveMQConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setPubSubDomain(true);
        //开启持久化订阅
        factory.setSubscriptionDurable(true);
        factory.setClientId("1002");
        return factory;
    }

    //异步接收消息
    @Bean
    public MyMessageListener messageListener() {
        return new MyMessageListener();
    }

    @Bean
    public MySessionAwareListener sessionAwareListener() {
        return new MySessionAwareListener();
    }

    @Bean
    public MyAdapterListener adapterListener() {
        return new MyAdapterListener();
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter() throws Exception {
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
        messageListenerAdapter.setDefaultListenerMethod("receiveMessage");
        messageListenerAdapter.setDelegate(adapterListener());
        messageListenerAdapter.setDefaultResponseQueueName(replyQueue().getQueueName());
        return messageListenerAdapter;
    }

    @Bean(name = "jmsContainer1")
    public SimpleMessageListenerContainer simpleListenerContainer() {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory("tcp://127.0.0.1:61616"));
        messageListenerContainer.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        messageListenerContainer.setMessageListener(messageListener());
        messageListenerContainer.setDestination(sendQueue());
        messageListenerContainer.setMessageSelector("listener=1");
        return messageListenerContainer;
    }

    @Bean(name = "jmsContainer2")
    public SimpleMessageListenerContainer sessionAwareListenerContainer() {
        SimpleMessageListenerContainer messageListenerContainer = new SimpleMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory("tcp://127.0.0.1:61616"));
        messageListenerContainer.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        messageListenerContainer.setMessageListener(sessionAwareListener());
        messageListenerContainer.setDestination(sendQueue());
        messageListenerContainer.setMessageSelector("listener=2");
        return messageListenerContainer;
    }

    @Bean(name = "jmsContainer3")
    public DefaultMessageListenerContainer messageListenerAdapterContainer() throws Exception {
        DefaultMessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer();
        messageListenerContainer.setConnectionFactory(connectionFactory("tcp://127.0.0.1:61616"));
        messageListenerContainer.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
        messageListenerContainer.setMessageListener(messageListenerAdapter());
        messageListenerContainer.setDestination(sendQueue());
        messageListenerContainer.setMessageSelector("listener=3");
        return messageListenerContainer;
    }
}
